package com.hexinfo.dmpro.sparing.service.impl;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import cn.hutool.core.io.file.FileWriter;
import cn.hutool.core.util.StrUtil;
import com.hexinfo.dmpro.common.utils.ConnExecuteUtil;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.component.comparison.service.ITableContrastService;
import com.hexinfo.dmpro.sparing.dto.DistCPDTO;
import com.hexinfo.dmpro.sparing.model.HdfsHeatBackup;
import com.hexinfo.dmpro.sparing.model.SyncCompleteLibrary;
import com.hexinfo.dmpro.sparing.model.UatSyncLogs;
import com.hexinfo.dmpro.sparing.service.BackupSourceService;
import com.hexinfo.dmpro.sparing.service.DistCPService;
import com.hexinfo.dmpro.sparing.service.HotStandbyLogTableService;
import com.hexinfo.dmpro.sparing.util.Logger;
import com.xqfunds.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.sql.Connection;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

@Service
@Slf4j
public class DistCPServiceImpl implements DistCPService {

    @Value("${hdfs.heatBackup}")
    private String heatBackup;

    @Lazy
    @Autowired
    private BackupSourceService backupSourceService;
    @Autowired
    private HotStandbyLogTableService hotStandbyLogTableService;

    //任务结束标志
    private Map<String, Boolean> endFlagMap = new ConcurrentHashMap<String, Boolean>();
    //修改结束状态
    public synchronized void updateEndFlag(String souName,Boolean state) {
        endFlagMap.put(souName, state);
    }
    //获取结束状态
    public Boolean getEndFlag(String souName) {
        return endFlagMap.get(souName);
    }
    //删除key
    public synchronized void deleteEndFlag(String souName) {
        endFlagMap.remove(souName);
    }

    @Override
    @Async("threadPoolTaskExecutor")
    public Future<Boolean> execShell(DistCPDTO distCPDTO, int sizes, String tableId) {
        String shell = heatBackup + " " + distCPDTO.getSouToTar();
        int i = execShellCode(shell,distCPDTO.getHeatlog(),distCPDTO.getTableLog());
        if (i != 0){
            logLog(distCPDTO.getHeatlog(),distCPDTO.getTableLog(),"DistCP命令：" + shell+"  --执行结果："+i+"（失败）","1");
            //表级日志分析记录修改--执行状态修改
            hotStandbyLogTableService.upState(tableId, ScanCommonConstants.ClusterName.TBSB.value);
            backupSourceService.updateTaskState(distCPDTO.getSourceDataName(),false);
            return new AsyncResult<>(false);
        }else {
            logLog(distCPDTO.getHeatlog(),distCPDTO.getTableLog(),"DistCP命令：" + shell+"  --执行结果："+i+"（成功）","3");
            return new AsyncResult<>(true);
        }
    }

    @Override
    public String execShellWgToUAT(DistCPDTO distCPDTO, UatSyncLogs uatSyncLogs) {
        String shell = heatBackup + " " + distCPDTO.getSouToTar();
        uatSyncLogs.setSql(shell);
        if(StrUtil.isNotBlank(uatSyncLogs.getLog())){
            uatSyncLogs.setLog(uatSyncLogs.getLog()+"\n["+DateUtil.now()+"]"+"【热备到UAT任务执行sql】"+shell);
        }else{
            uatSyncLogs.setLog("["+DateUtil.now()+"]【热备到UAT任务执行sql】"+shell);
        }
        String err = execShellCodeWgToUat(shell);
        if(err.contains("失败")){
            uatSyncLogs.setLog(uatSyncLogs.getLog()+"\n["+DateUtil.now()+"]"+err+"\n["+DateUtil.now()+"]【热备到UAT任务执行状态】失败");
        }else{
            uatSyncLogs.setLog(uatSyncLogs.getLog()+"\n["+DateUtil.now()+"]【热备到UAT任务执行状态】成功");
        }
        return err;
    }

    @Override
    public String execShellDatabase(DistCPDTO distCPDTO, SyncCompleteLibrary libraryRb) {
        String shell = heatBackup + " " + distCPDTO.getSouToTar();
        String jg = execShellCodeDatabase(shell);
        if (jg.contains("成功")){
//            Logger.log("DistCP命令：" + shell+"  --执行结果：（成功）");
            libraryRb.setLog(libraryRb.getLog()+"\n["+DateUtil.now()+"]【DistCP命令】"+shell+"  --执行结果：（成功）");
        }else {
//            Logger.log("DistCP命令：" + shell+"  --执行结果：（失败）\n"+jg);
            libraryRb.setLog(libraryRb.getLog()+"\n["+DateUtil.now()+"]【DistCP命令】"+shell+"  --执行结果：（失败）\n");
            return jg;
        }
        return jg;
    }

    /**
     * 单表完成修改状态、日志
     */
    @Override
    public void souTableIsFinish(String sourceDataName,String tableId,String souDbTbName,FileAppender heatlog,FileAppender tableLog,boolean state){
        logLog(heatlog,tableLog,"【同步完成】--"+souDbTbName,"1");
        logLog(heatlog,tableLog,"-------------------------【数据同步结束】","3");
        logLog(heatlog,tableLog,"","4"); //表日志刷进日志中
        logLog(heatlog,tableLog,"","2"); //总日志刷进日志中
        //表级日志分析记录修改--用时详情修改
        hotStandbyLogTableService.upHotLog(tableId,DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"\n");
    }

    /**
     * 执行shell脚本，只返回执行结果
     * @param pathOrCommand 脚本路径或者命令
     * @return
     */
    private int execShellCode(String pathOrCommand, FileAppender heatlog, FileAppender tableLog) {
        int retryCount = 0; // 当前重试次数
        int exitValue = 999; //SH脚本执行状态码
        while (retryCount < 3) {
            try {
                // 调用需要重试的方法
                exitValue = execShellCS(pathOrCommand, heatlog, tableLog);
                // 方法调用成功，直接退出循环
                if(exitValue!=999){
                    break;
                }
            } catch (Exception e) {
                // 方法调用出现异常，进行重试
                System.out.println("方法调用出现异常，进行第 " + (retryCount + 1) + " 次重试");
                // 等待一段时间，可以使用 Thread.sleep() 方法来实现
                // 在实际应用中，根据具体情况来决定等待时间
                try {
                    Thread.sleep(1000); // 等待1秒
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
                retryCount++;
            }
        }
        return exitValue;
    }

    private int execShellCS(String pathOrCommand, FileAppender heatlog, FileAppender tableLog) {
        int exitValue = 0;
        String psErr = "";
        try {
            Process ps = Runtime.getRuntime().exec(pathOrCommand);
            exitValue = ps.waitFor();
            // 获取脚本错误输出
            String line;
            BufferedReader errorReader = new BufferedReader(new InputStreamReader(ps.getErrorStream()));
            while ((line = errorReader.readLine()) != null) {
                psErr += line+"\n";
            }
            if (exitValue != 0 || (StrUtil.isNotBlank(psErr)&&psErr.contains("ERROR"))){
                logLog(heatlog,tableLog,"【按源执行DistCP错误】shell："+pathOrCommand+psErr+"---执行异常：","1");
                return 999;
            }
        } catch (Exception e) {
            log.error("执行DistCP脚本失败命令：{}，执行DistCP脚本失败报错：{}",pathOrCommand,e);
            logLog(heatlog,tableLog,"【按源执行DistCP错误】shell："+pathOrCommand+psErr+"---执行异常："+e,"1");
            return 999;
        }
        return exitValue;
    }

    private String execShellCSDatabase(String pathOrCommand) {
        String exitValue = "成功";
        String psErr = "";
        int jg;
        try {
            Process ps = Runtime.getRuntime().exec(pathOrCommand);
            jg =  ps.waitFor();
            // 获取脚本错误输出
            String line;
            BufferedReader errorReader = new BufferedReader(new InputStreamReader(ps.getErrorStream()));
            while ((line = errorReader.readLine()) != null) {
                psErr += line+"\n";
            }
            if (jg != 0 || (StrUtil.isNotBlank(psErr)&&psErr.contains("ERROR"))){
                return "执行DistCP脚本失败命令：{"+pathOrCommand+"}，执行DistCP脚本失败报错：{"+psErr+"}";
            }
        } catch (Exception e) {
            return "执行DistCP脚本失败命令：{"+pathOrCommand+"}，执行DistCP脚本失败报错：{"+psErr+e.getMessage()+"}";

        }
        return exitValue;
    }

    /**
     * 执行shell脚本，只返回执行结果 热备到UAT
     * @param pathOrCommand 脚本路径或者命令
     * @return
     */
    private String execShellCodeDatabase(String pathOrCommand) {
        int retryCount = 0; // 当前重试次数
        String exitValue = "成功"; //SH脚本执行状态码
        while (retryCount < 3) {
            try {
                // 调用需要重试的方法
                exitValue = execShellCSDatabase(pathOrCommand);
                // 方法调用成功，直接退出循环
                if(exitValue.equals("成功")){
                    break;
                }
            } catch (Exception e) {
                // 方法调用出现异常，进行重试
                System.out.println("方法调用出现异常，进行第 " + (retryCount + 1) + " 次重试");
                // 等待一段时间，可以使用 Thread.sleep() 方法来实现
                // 在实际应用中，根据具体情况来决定等待时间
                try {
                    Thread.sleep(1000); // 等待1秒
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
                retryCount++;
            }
        }
        return exitValue;
    }

    /**
     * 执行shell脚本，只返回执行结果 热备到UAT
     * @param pathOrCommand 脚本路径或者命令
     * @return
     */
    private String execShellCodeWgToUat(String pathOrCommand) {
        int retryCount = 0; // 当前重试次数
        String exitValue = "成功"; //SH脚本执行状态码
        while (retryCount < 3) {
            try {
                // 调用需要重试的方法
                exitValue = execShellCSDatabase(pathOrCommand);
                // 方法调用成功，直接退出循环
                if(exitValue.equals("成功")){
                    break;
                }
            } catch (Exception e) {
                // 方法调用出现异常，进行重试
                System.out.println("方法调用出现异常，进行第 " + (retryCount + 1) + " 次重试");
                // 等待一段时间，可以使用 Thread.sleep() 方法来实现
                // 在实际应用中，根据具体情况来决定等待时间
                try {
                    Thread.sleep(1000); // 等待1秒
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
                retryCount++;
            }
        }
        return exitValue;
    }

    /**
     * 加锁，防止并发问题
     * @param heatlog
     * @param tableLog
     * @param logStr
     * @param str
     */
    private synchronized void logLog(FileAppender heatlog, FileAppender tableLog, String logStr ,String str) {
        if (str.equals("1")){
            heatlog.append(logTime(logStr));
            tableLog.append(logTime(logStr));
        }else if (str.equals("2")){
            heatlog.flush();     //总日志刷进日志中
        }else if (str.equals("3")){
            tableLog.append(logTime(logStr));
        }else if (str.equals("4")){
            tableLog.flush();     //表日志刷进日志中
        }
    }

    /**
     * 日志加日期时间
     * @param str
     * @return
     */
    private String logTime(String str) {
        return "["+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }

}
